package cn.itcast.tags.models.statistics

import cn.itcast.tags.models.{AbstractModel, ModelType}
import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.types.{IntegerType, StringType}
import org.apache.spark.sql.{DataFrame, Dataset, Row}

class AgeRangeModel2 extends AbstractModel("AgeRangeModel2",ModelType.STATISTICS){
  override def doTag(businessDF: DataFrame, tagDF: DataFrame): DataFrame = {
    import businessDF.sparkSession.implicits._
    import org.apache.spark.sql.functions._

    /**
     * root
     * |-- id: string (nullable = true)
     * |-- birthday: string (nullable = true)
     *
     * +---+----------+
     * |id |birthday  |
     * +---+----------+
     * |1  |1992-05-31|
     */


    /**
     * +---+----+-----------------+-----+
     * |id |name|rule             |level|
     * +---+----+-----------------+-----+
     * |333|50后 |19500101-19591231|5    |
     */
    /**
     * +-----+--------+--------+
     * |tagId|start   |end     |
     * +-----+--------+--------+
     * |333  |19500101|19591231|
     */
    val rule_to_tag: UserDefinedFunction = udf(
      (rule: String) => {
        val Array(start, end) = rule.split("-").map(_.toInt)
        (start, end)
      }
    )
    val attrTagModel: DataFrame = tagDF
      .filter($"level" === 5)
        .select($"id",
          rule_to_tag($"rule").as("rules")
        )
        .select(
          $"id".as("tagId"),
          $"rules._1".as("start"),
          $"rules._2".as("end")
        )
    attrTagModel.printSchema()
    attrTagModel.show(10,false)

    val birthdayDF: DataFrame = businessDF.select(
      $"id",
      regexp_replace($"birthday", "-", "").cast(IntegerType).as("bornday")
    )

    birthdayDF.printSchema()
    birthdayDF.show(10,false)

    /**
     * root
     * |-- id: string (nullable = true)
     * |-- bornday: integer (nullable = true)
     * |-- tagId: long (nullable = false)
     * |-- start: integer (nullable = true)
     * |-- end: integer (nullable = true)
     */
    val modelDF: DataFrame = birthdayDF.join(attrTagModel)
        .where(
          birthdayDF("bornday").between(attrTagModel("start"),attrTagModel("end"))
        )
        .select(
          $"id".as("uid"),
          $"tagId".cast(StringType)
        )



    modelDF.printSchema()
    modelDF.show(10,false)
    println(modelDF.count())
    modelDF
  }
}

object AgeRangeModel2{
  def main(args: Array[String]): Unit = {
    val ageRangeModel = new AgeRangeModel2()
    ageRangeModel.executeModel(332L,false)
  }
}
